package vertex;

import static java.lang.System.err;
import static java.lang.System.exit;
import static java.lang.Math.sqrt;

import java.awt.geom.Point2D;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.StringTokenizer;
import system.Combiner;
import system.Message;
import system.Vertex;
import system.aggregators.IntegerSumAggregator;

/**
 *
 * The value associated with each vertex, v, is: 
 * 1. the minimum distance to vertex v
 * 2. the vertexId of vertex u, where edge (u, v) is on a shortest path to v.
 * Since these 2 items constitute a Message, vertexValue is a Message.
 * 
 * @author Pete Cappello
 */
public class EuclideanShortestPathVertex extends Vertex<Point2D.Double, Double>
{
    public EuclideanShortestPathVertex( Point2D.Double vertexId, Map<Object, Point2D.Double> outEdgeMap, Combiner combiner )
    {
        super( vertexId, outEdgeMap, combiner );
        setVertexValue( new Message( vertexId, Double.MAX_VALUE ) );
    }
    
    public EuclideanShortestPathVertex() {}
    
    public Vertex make( String line, Combiner combiner )
    {
        StringTokenizer stringTokenizer = new StringTokenizer( line );
        if ( ! stringTokenizer.hasMoreTokens() )
        {
            err.println( "Vertex.make: Empty lines are not allowed." );
            exit( 1 );
        }
        double x  = Double.parseDouble( stringTokenizer.nextToken() );
        double y  = Double.parseDouble( stringTokenizer.nextToken() );
        Point2D.Double vertexId = new Point2D.Double( x, y);
        Map<Object, Point2D.Double> outEdgeMap = new HashMap<Object, Point2D.Double>();
        while( stringTokenizer.hasMoreTokens() )
        { 
            x  = Double.parseDouble( stringTokenizer.nextToken() );
            y  = Double.parseDouble( stringTokenizer.nextToken() );
            Point2D.Double target = new Point2D.Double( x, y );
            outEdgeMap.put( target, target );
        }        
        return new EuclideanShortestPathVertex( vertexId, outEdgeMap, combiner );
    }  

    @Override
    public void compute() 
    {
        // compute currently known minimum distance from source to me
        Double minDistance = isSource() ? 0.0 : Double.MAX_VALUE;
        Message<Double> minDistanceMessage = new Message<Double>( getVertexId(), minDistance );
        Iterator<Message<Double>> messageSetIterator = getMessageIterator();
        while ( messageSetIterator.hasNext() )
        {
            Message<Double> message = messageSetIterator.next();
            if ( message.getMessageValue() < minDistanceMessage.getMessageValue() )
            {
                minDistanceMessage = message;
            }
        }
        if ( minDistanceMessage.getMessageValue() < ((Message<Double>) getVertexValue() ).getMessageValue() )
        {
            // aggregate number of messages sent in this step & in this problem
//            setOutputStepAggregator(    new IntegerSumAggregator( getOutEdgeMapSize() ) );
            setOutputProblemAggregator( new IntegerSumAggregator( getOutEdgeMapSize() ) );
            
            // there is a new shorter path from the source to me
            setVertexValue( minDistanceMessage ); // update my value: the shortest path to me
            
            // To each target vertex: The shortest known path length to you through me just got shorter 
//            for ( Iterator<Point2D.Double> outEdgeIterator = getOutEdgeValues(); outEdgeIterator.hasNext(); )
            for ( Point2D.Double targetVertexId : getOutEdgeValues() )            
            {
                double edgeValue = distance( targetVertexId );
                Message<Double> message = new Message<Double>( getVertexId(), minDistanceMessage.getMessageValue() + edgeValue );
//                out.println( getSuperStep() + ": " + getVertexId() + "  sending: " + targetVertexId + " edgeValue: " + edgeValue 
//                        + " path length - edgeValue: " + minDistanceMessage.getMessageValue());
                sendMessage( targetVertexId, message );
            }
        }
        else
        {
            // aggregate number of messages sent in this step
//            setOutputStepAggregator(    new IntegerSumAggregator( 0 ) );
            setOutputProblemAggregator( new IntegerSumAggregator( 0 ) );
        }       
        
//        out.println("EuclideanShortestPathVertex.compute: Step: " + getSuperStep() + ": vertexId: " + getVertexId() + " from: "
//                + ((Message<Double>) getVertexValue()).getVertexId() 
//                + " length: " + ((Message<Double>)getVertexValue()).getMessageValue() );
        
        /* This vote will be overturned, if during this step, a vertex for whom 
         * I am a target vertex discovered a shorter path to itself, 
         * in which case, it will send me a message.
         */   
        voteToHalt(); 
    }

    @Override
    public String output() 
    {
        StringBuffer stringBuffer = new StringBuffer();
        stringBuffer.append( (Point2D.Double) getVertexId() );
        stringBuffer.append( " : ");
        stringBuffer.append( (Point2D.Double) ((Message)  getVertexValue() ).getVertexId() );
        stringBuffer.append( " --- ");
        stringBuffer.append( (Double) ((Message)  getVertexValue() ).getMessageValue() );
        return new String( stringBuffer );
    }
    
    boolean isSource()
    {
        Point2D.Double vertex = (Point2D.Double) getVertexId();
        return ( vertex.getX() == 0.0 && vertex.getY() == 0.0 ) ? true : false; 
    }
    
    private double distance( Point2D.Double targetVertexId )
    {
        double x1 = ((Point2D.Double) getVertexId()).getX();
        double y1 = ((Point2D.Double) getVertexId()).getY();
        double x2 = targetVertexId.getX();
        double y2 = targetVertexId.getY();
        double deltaX = x1 - x2;
        double deltaY = y1 - y2;
        return sqrt( deltaX * deltaX + deltaY * deltaY );
    }
}
